package MountainMq.config;

import MountainMq.service.TransMessageService;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.RabbitListenerContainerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.amqp.RabbitAutoConfiguration;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
@ConditionalOnProperty("mountainmq.enable")
@Slf4j
// 继承了RabbitAutoConfiguration，就可以控制是否开局自动加载一些配置
public class MountainRabbitConfig extends RabbitAutoConfiguration{
    @Value("${mountainmq.host}")
    private String host;
    @Value("${mountainmq.port}")
    private Integer port;
    @Value("${mountainmq.username}")
    private String username;
    @Value("${mountainmq.password}")
    private String password;
    @Value("${mountainmq.vhost}")
    private String vhost;
    @Value("${mountainmq.resendTimes}")
    private int resendTimes;

    @Autowired
    private TransMessageService transMessageService;

    // jackson序列化器
    @Bean
    public ObjectMapper objectMapper(){
        return new ObjectMapper();
    }

    @Bean
    public ConnectionFactory connectionFactory(){
        CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
        connectionFactory.setHost(host);
        connectionFactory.setPort(port);
        connectionFactory.setUsername(username);
        connectionFactory.setPassword(password);
        connectionFactory.setVirtualHost(vhost);
        // 发送者确认 确认是异步的，所以需要知道是哪条消息，那么就要CORRELATED参数
        connectionFactory.setPublisherConfirmType(CachingConnectionFactory.ConfirmType.CORRELATED);
        // 消息返回 消息无法路由的时候 打开消息确认回调功能，不设置的话就不会触发消息确认回调
        connectionFactory.setPublisherReturns(true);
        // 由于懒加载机制，这里我们手动调用以下，RabbitAdmin才会自动去创建队列之类的
        connectionFactory.createConnection();
        return connectionFactory;
    }

    @Bean
    public RabbitAdmin rabbitAdmin(ConnectionFactory connectionFactory){
        RabbitAdmin admin = new RabbitAdmin(connectionFactory);
        // 启动时候自动搜索spring容器中，有关queue等，自动去声明
        admin.setAutoStartup(true);
        return admin;
    }

    // 下面这两个工厂都是给@RabbitListener用的
    // 默认无事务工厂  继承了RabbitAutoConfiguration后就默认有个名为rabbitListenerContainerFactory的bean
    // 继续用这个的话，名字就不能是rabbitListenerContainerFactory，得改个名
//    @Bean
//    public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory(ConnectionFactory connectionFactory){
//        SimpleRabbitListenerContainerFactory factory
//                = new SimpleRabbitListenerContainerFactory();
//        factory.setConnectionFactory(connectionFactory());
//        // 下面两个参数其实也可以从配置文件中获取，但是现在为了方便暂时先写死
//        factory.setConcurrentConsumers(3);
//        factory.setMaxConcurrentConsumers(10);
//        factory.setAutoStartup(false);
//        return factory;
//    }

    // 事务工厂 也就是需要手动ACK
    @Bean
    public RabbitListenerContainerFactory commonRabbitListenerContainerFactory(ConnectionFactory connectionFactory){
        SimpleRabbitListenerContainerFactory factory
                = new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory());
        // 下面两个参数其实也可以从配置文件中获取，但是现在为了方便暂时先写死
        factory.setConcurrentConsumers(3);
        factory.setMaxConcurrentConsumers(10);
        // 手动确认
        factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);
        return factory;
    }

    // 自定义RabbitTemplate
    @Bean
    public RabbitTemplate customRabbitTemplate(ConnectionFactory connectionFactory){
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        // 设置托管 为true才会处理无法路由的消息 否则直接丢弃
        rabbitTemplate.setMandatory(true);

        // 设置消息返回 消息发送成功的回调当消息进入Exchange交换器时就进入回调,不管是否进入队列
        // 消息可能是 确认 或 不确认
        rabbitTemplate.setConfirmCallback(((correlationData, ack, cause) -> {
            log.info("correlationData:{},ack:{},cause:{}",correlationData,ack,cause);
            if(ack && null != correlationData){ // 有该交换机，且有correlationData(messageID)
                String messageId = correlationData.getId();
                log.info("消息已经正确投递到交换机，id:{}",messageId);
                // 消息发送成功，直接把该消息从数据库中硬删除 感觉后序可以改成软删除
                transMessageService.messageSendSuccess(messageId);
            }else{ // 没有该交换机，或者投递交换机失败
                log.error("消息投递至交换机失败，correlationData:{}",correlationData);
            }
        }));

        // 设置消息回调 消息返回是一定异常，因为消息无法路由到队列 说明没有该队列 消息找不到队列
        rabbitTemplate.setReturnCallback((message,replyCode,replyText,exchange,routingKey)->{
            log.error("消息无法路由！message:{},replyCode:{},replyText:{},exchange:{},routingKey:{}",
                    message,replyCode,replyText,exchange,routingKey);
            // messageBody
            byte[] body = message.getBody();
            // 这里拿到的是消息本身的ID，在消息发送的时候我们设置过了
            String messageId = message.getMessageProperties().getMessageId();
            // 消息返回就再次持久化到库中(原有逻辑，跟硬删除绑定在一起，软删除不能这样)
            transMessageService.messageSendReturn(messageId,exchange
                    ,routingKey,new String(body));

            // 个人见解：这里我觉得不应该重新持久化到库中，应该直接软删除，不然会陷入死循环的
//            transMessageService.messageSendSuccess(messageId);
        });
        return rabbitTemplate;
    }

}
